package com.jzl.canal_consumer.etl.load.mysql.impl;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.jzl.canal_consumer.bean.DdlEntry;
import com.jzl.canal_consumer.bean.DmlEntry;
import com.jzl.canal_consumer.bean.MergeSqlEntry;
import com.jzl.canal_consumer.bean.TableBatchMergeSqlEntry;
import com.jzl.canal_consumer.etl.load.mysql.Loader;
import com.jzl.canal_consumer.etl.load.mysql.SqlExecutor;
import com.jzl.canal_consumer.util.MessageEntryUtil;
import com.jzl.util.LoggerUtil;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

/**
 * @author ZhenWuWang
 */
public class MysqlLoadImpl implements Loader<TableBatchMergeSqlEntry>
{
    private final static int RETRY_TIMES = 3;

    private final static MessageEntryUtil.ExceptionStrategy EXCEPTION_STRATEGY = MessageEntryUtil.ExceptionStrategy.RETRY;

    private final SqlExecutor sqlExecutor;

    private final MessageEntryUtil messageEntryUtil;

    public MysqlLoadImpl(SqlExecutor sqlExecutor, MessageEntryUtil messageEntryUtil)
    {
        this.sqlExecutor = sqlExecutor;
        this.messageEntryUtil = messageEntryUtil;
    }

    @Override
    public void load(List<TableBatchMergeSqlEntry> list)
    {
        if (list == null || list.size() == 0)
        {
            return;
        }

        for (TableBatchMergeSqlEntry tableBatchMergeSqlEntry : list)
        {
            processTableBatchMergeSqlEntry(tableBatchMergeSqlEntry);
        }
    }


    private void processTableBatchMergeSqlEntry(TableBatchMergeSqlEntry tableBatchMergeSqlEntry)
    {
        String tableName = tableBatchMergeSqlEntry.getTableName();
        LoggerUtil.info("tableName:" + tableName);
//        List<CanalEntry.Entry> entryList = tableBatchMergeSqlEntry.getTransactionEntries();
//        if (entryList != null && entryList.size() > 0)
//        {
//            messageEntryUtil.printEntry(entryList);
//        }
        List<MergeSqlEntry> mergeSqlEntries = tableBatchMergeSqlEntry.getMergeSqlEntries();
        for (MergeSqlEntry sqlEntry : mergeSqlEntries)
        {
            processMergeSqlEntry(tableName, sqlEntry);
        }
    }

    private void processMergeSqlEntry(String tableName, MergeSqlEntry sqlEntry)
    {
        DdlEntry ddlEntry = sqlEntry.getDdlEntry();
        if (ddlEntry != null)
        {
            if (!retryProcess(ddlEntry, this::processDdl))
            {
                LoggerUtil.error(String.format("执行DDL语句失败:\ntableName:%s\nDDL SQL:%s",
                        tableName, ddlEntry.getSql()));
            }
        }

        Map<CanalEntry.EventType, List<DmlEntry>> map = sqlEntry.merge();
        map.forEach((eventType, dmlEntries) -> retryProcess(dmlEntries, entries -> processDml(eventType, entries)));
    }

    private void processDml(CanalEntry.EventType eventType, List<DmlEntry> dmlEntries)
    {
        CanalEntry.Header header = dmlEntries.get(0).getHeader();
        List<List<CanalEntry.Column>> columns = new ArrayList<>();
        for (DmlEntry dmlEntry : dmlEntries)
        {
            CanalEntry.RowData rowData = dmlEntry.getRowData();
            List<CanalEntry.Column> cl = null;
            if (eventType == CanalEntry.EventType.INSERT || eventType == CanalEntry.EventType.UPDATE)
            {
                cl = rowData.getAfterColumnsList();
            }
            else if (eventType == CanalEntry.EventType.DELETE)
            {
                cl = rowData.getBeforeColumnsList();
            }
            if (cl != null && cl.size() > 0)
            {
                columns.add(cl);
            }
        }
        if (columns.size() > 0)
        {
            sqlExecutor.batchExecute(header, columns);
        }
    }

    private void processDdl(DdlEntry ddlEntry)
    {
        messageEntryUtil.processDDL(ddlEntry.getHeader(), ddlEntry.getEventType(), ddlEntry.getSql());
    }

    private <T> boolean retryProcess(T t, Consumer<T> consumer)
    {
        int times = 0;
        boolean success = false;
        String exceptionMsg = "";
        while (!success)
        {
            if (times > 0)
            {
                //1:retry，重试，重试默认为3次，由retryTimes参数决定，如果重试次数达到阈值，则跳过，并且记录日志。
                //2:ignore,直接忽略，不重试，记录日志。
                if (EXCEPTION_STRATEGY == MessageEntryUtil.ExceptionStrategy.RETRY)
                {
                    if (times >= RETRY_TIMES)
                    {
                        break;
                    }
                }
                else
                {
                    break;
                }
            }
            try
            {
                consumer.accept(t);
                success = true;
            } catch (Exception e)
            {
                times++;
                exceptionMsg = e.toString();
            }
        }
        if (!success)
        {
            LoggerUtil.error(String.format("Process event has an error{times: %s},{data:%s},{exceptionMsg:%s}",
                    times, t.toString(), exceptionMsg));
        }
        return success;
    }
}
